/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.snda.lidejia.jobflow.task;

import com.snda.lidejia.jobflow.exception.JobFlowException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 *
 * @author lidejia
 */
public class MultiThreadTask extends AbstractTask {

    private int threadNumber = 0;
    private Class<? extends TaskHandler> taskHandlerClass = null;

    public MultiThreadTask(Class<? extends TaskHandler> taskHandlerClass, int threadNumber) {
        this.taskHandlerClass = taskHandlerClass;
        this.threadNumber = threadNumber;
    }

    public TaskData run(TaskData inputs) throws JobFlowException {
        try {
            // the max number of thread should be the same as the input data size.
            threadNumber = Math.min(threadNumber, inputs.size());
            // thread pool
            ExecutorService es = Executors.newFixedThreadPool(threadNumber);
            // split the input data.
            List<TaskData> splitedInputs = inputs.split(threadNumber);
            // run each split in a thread
            List<Future<TaskData>> taskFutures = new ArrayList<Future<TaskData>>(threadNumber);
            for (final TaskData input : splitedInputs) {
                Future<TaskData> future = es.submit(new Callable<TaskData>() {

                    public TaskData call() throws Exception {
                        Task task = new SingleThreadTask(taskHandlerClass);
                        return task.run(input);
                    }
                });
                taskFutures.add(future);
            }
            // merge the result of each thread
            TaskData output = new TaskData();
            for (Future<TaskData> future : taskFutures) {
                output.merge(future.get());
            }
            return output;
        } catch (Exception ex) {
            throw new JobFlowException(ex);
        }

    }
}
